package com.bff.gaia.streaming.api.xjoin.operators;/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import com.bff.gaia.api.common.typeinfo.TypeInformation;
import com.bff.gaia.api.java.typeutils.RowTypeInfo;
import com.bff.gaia.streaming.api.datastream.DataStream;
import com.bff.gaia.streaming.api.datastream.SingleOutputStreamOperator;
import com.bff.gaia.streaming.api.xjoin.core.enums.ECacheType;
import com.bff.gaia.streaming.api.xjoin.core.enums.EJoinType;
import com.bff.gaia.streaming.api.xjoin.core.side.FieldInfo;
import com.bff.gaia.streaming.api.xjoin.core.side.JoinInfo;
import com.bff.gaia.streaming.api.xjoin.core.side.SideTableInfo;
import com.bff.gaia.types.Row;

import java.util.ArrayList;
import java.util.List;

/**
 * Project Name: gaia-parent
 * Description:
 * Data: 2019/7/23 16:31
 * Company: NEU
 *
 * @author tiger
 * @version v1.0
 */
public class XjoinStreamV3 {
	private DataStream<Row> left;// 源数据流
	private SideTableInfo right; //维表
	private List<FieldInfo> outFieldInfoList; //输出列
	private JoinInfo joinInfo; //连接辅助参数


	public XjoinStreamV3(DataStream left, SideTableInfo right) {
		this.left = left;
		this.right = right;
	}

	public ProjectLeft projectLeftKey(int... leftKeyPosition) {
		return new ProjectLeft(leftKeyPosition);
	}

//	public ProjectLeft projectLeftKey(String... leftKeyName) {
//		return new ProjectLeft(leftKeyName);
//	}

	public class ProjectLeft {
		int[] leftKeyPosition;
		int[] leftFieldPosition;
		RowTypeInfo leftKeyType;

		ProjectLeft(int... leftKeyPosition) {

			this.leftKeyPosition = leftKeyPosition;
//			System.out.println(left.getType().getArity());
			this.leftKeyType = RowTypeInfo.projectFields((RowTypeInfo) left.getType(), leftKeyPosition);
			leftFieldPosition = projectPosition(left.getType().getArity());
		}

//		ProjectLeft(String... leftKeyName) {
//			int[] tempKey = new int[leftKeyName.length];
//			for (int i = 0; i < tempKey.length; i++) {
//				tempKey[i] = ((RowTypeInfo)left.getType()).getFieldIndex(leftKeyName[i]);
//			}
//			this.leftKeyPosition = tempKey;
//
//			this.leftKeyType = RowTypeInfo.projectFields((RowTypeInfo)left.getType(),leftKeyPosition);
//
//			leftFieldPosition = projectPosition(left.getType().getArity());
//		}

		public ProjectLeft projectLeftField(int... leftFieldPosition) {

			if (isSubSets(leftFieldPosition, leftKeyPosition)) {
				throw new IllegalArgumentException("project fields must have key");
			}

			this.leftFieldPosition = leftFieldPosition;


			return this;
		}

		public ProjectRight projectRightKey(int... rightKeyPosition) {
			if (leftKeyPosition.length != rightKeyPosition.length) {
				throw new IllegalArgumentException("wrong key num");
			}
			return new ProjectRight(rightKeyPosition);
		}

//		public ProjectRight projectRightKey(String... rightKeyName) {
//			if (leftKeyPosition.length != rightKeyName.length) {
//				throw new IllegalArgumentException("wrong key num");
//			}
//			return new ProjectRight(rightKeyName);
//		}

		public class ProjectRight {
			int[] rightKeyPosition;
			int[] rightFieldPosition;
			RowTypeInfo rightKeyType;

			ProjectRight(int... rightKeyPosition) {
				this.rightKeyPosition = rightKeyPosition;

				rightKeyType = projectFields(right.getFieldTypeInfomationList(), right.getFieldList(), rightKeyPosition);

				rightFieldPosition = projectPosition(right.getFieldList().size());
			}

			public ProjectRight projectRightField(int... rightFieldPosition) {
				if (isSubSets(rightFieldPosition, rightKeyPosition)) {
					throw new IllegalArgumentException("project fields must have key");
				}
				this.rightFieldPosition = rightFieldPosition;
				return this;
			}

//			ProjectRight(String... rightKeyName) {
//				int[] tempKey = new int[rightKeyName.length];
//				for (int i = 0; i < tempKey.length; i++) {
//					tempKey[i] = right.getFieldList().indexOf(rightKeyName[i]);
//				}
//				this.rightKeyPosition = tempKey;
//
//				rightKeyType = projectFields(right.getFieldTypeInfomationList(),right.getFieldList(),rightKeyPosition);
//
//				rightFieldPosition = projectPosition(right.getFieldList().size());
//			}

			public SetArgs setCacheType(ECacheType cacheType) {
				return new SetArgs(cacheType);
			}

			public class SetArgs {
				private String cacheType = "LRU";//None or LRU or ALL

				private int cacheSize = 10000;

				private long cacheTimeout = 60 * 1000;//

				private int asyncCapacity = 100;

				private int parallelism = 1;

				SetArgs(ECacheType cacheType) {
					this.cacheType = cacheType.name();
				}

				public SetArgs setCacheSize(int cacheSize) {
					this.cacheSize = cacheSize;
					return this;
				}

				public SetArgs setCacheTimeout(long cacheTimeout) {
					this.cacheTimeout = cacheTimeout;
					return this;
				}

				public SetArgs setAsyncCapacity(int asyncCapacity) {
					this.asyncCapacity = asyncCapacity;
					return this;
				}

				public SetArgs setParallelism(int parallelism) {
					this.parallelism = parallelism;
					return this;
				}

				public SingleOutputStreamOperator<Row> apply(EJoinType joinType) throws Exception {
					// 构建缓存信息
					right.setCacheType(cacheType);
					right.setCacheSize(cacheSize);
					right.setCacheTimeout(cacheTimeout);
					right.setAsyncCapacity(asyncCapacity);
					right.setParallelism(parallelism);


					ArrayList<TypeInformation<?>> typeInformations = new ArrayList<>();
					ArrayList<String> strings = new ArrayList<>();

					//构建输出数据结构
					outFieldInfoList = new ArrayList<>();
					for (int i : leftFieldPosition) {
						FieldInfo fieldInfo = new FieldInfo();
						for (int ii : leftKeyPosition) {
							if (i == ii) {
								fieldInfo.setJoinKey(true);
								break;
							}
						}
						fieldInfo.setTable("stream");
						fieldInfo.setFieldName(((RowTypeInfo) left.getType()).getFieldNames()[i]);
						fieldInfo.setTypeInformation(((RowTypeInfo) left.getType()).getTypeAt(i));
						outFieldInfoList.add(fieldInfo);

						typeInformations.add(fieldInfo.getTypeInformation());
						strings.add(fieldInfo.getFieldName());
					}
					for (int j : rightFieldPosition) {
						FieldInfo fieldInfo = new FieldInfo();
						for (int jj : rightKeyPosition) {
							if (j == jj) {
								fieldInfo.setJoinKey(true);
								break;
							}
						}
						fieldInfo.setTable(right.getName());
						fieldInfo.setFieldName(right.getFieldList().get(j));
						fieldInfo.setTypeInformation(right.getFieldTypeInfomationList().get(j));
						if (!fieldInfo.isJoinKey()) {
							typeInformations.add(fieldInfo.getTypeInformation());
							strings.add(fieldInfo.getFieldName());
						}
						outFieldInfoList.add(fieldInfo);
					}

					//构建连接信息
					JoinInfo joinInfo = new JoinInfo();
					joinInfo.setLeftTableName("stream");
					joinInfo.setLeftTableAlias("stream");
					joinInfo.setLeftIsSideTable(false);
					joinInfo.setRightTableName(right.getName());
					joinInfo.setRightTableAlias(right.getName());
					joinInfo.setRightIsSideTable(true);
					joinInfo.setJoinType(joinType);


					//选择缓存模式
					if (cacheType.equals(ECacheType.LRU.name())) {
						return XjoinAsyncOperator
							.getSideJoinDataStream(left, right.getType(), ((RowTypeInfo) left.getType()), joinInfo, outFieldInfoList, right)
							.returns(
								new RowTypeInfo(
									typeInformations.toArray(new TypeInformation[typeInformations.size()]),
									strings.toArray(new String[strings.size()])
								)
							);

					} else {
						return XjoinAllOperator
							.getSideJoinDataStream(left, right.getType(), ((RowTypeInfo) left.getType()), joinInfo, outFieldInfoList, right)
							.returns(
								new RowTypeInfo(
									typeInformations.toArray(new TypeInformation[typeInformations.size()]),
									strings.toArray(new String[strings.size()])
								)
							);
					}
				}
			}

		}
	}

//	public static TupleTypeInfo projectFields(TupleTypeInfo typeInfo, int[] fieldMapping) {
//		TypeInformation[] fieldTypes = new TypeInformation[fieldMapping.length];
//		String[] fieldNames = new String[fieldMapping.length];
//		for (int i = 0; i < fieldMapping.length; i++) {
//			fieldTypes[i] = typeInfo.getTypeAt(fieldMapping[i]);
//		}
//		return new TupleTypeInfo(fieldTypes);
//	}

	public static RowTypeInfo projectFields(List<TypeInformation> typeInfo, List<String> fieldName, int[] fieldMapping) {
		TypeInformation[] fieldTypes = new TypeInformation[fieldMapping.length];
		String[] fieldNames = new String[fieldMapping.length];
		for (int i = 0; i < fieldMapping.length; i++) {
			fieldTypes[i] = typeInfo.get(fieldMapping[i]);
			fieldNames[i] = fieldName.get(fieldMapping[i]);
		}
		return new RowTypeInfo(fieldTypes, fieldNames);
	}

	private static int[] projectPosition(int l) {
		int[] temp = new int[l];
		for (int i = 0; i < l; i++) {
			temp[i] = i;
		}
		return temp;
	}

	private static boolean isSubSets(int[] father, int[] child) {
		for (int i : child) {
			boolean isSubSet = false;
			for (int j : father) {
				if (j == i) {
					isSubSet = true;
					break;
				}
			}
			if (!isSubSet) {
				return true;
			}
		}
		return false;
	}

}